import re
from collections import namedtuple, OrderedDict
from json import loads

import pandas as pd
from bs4 import BeautifulSoup
from requests import request as http
from sqlalchemy.orm import Session

from app.bigdata.models.kudu_table_info import KuduTableInfo
from app.bigdata.utils import unit_convert
from setting import config
from core.logger import logger


class KuduInfo(object):
    def __init__(self, protocol: str = 'http', host: str = config.KUDU_MASTER_HOST, port: int = 8051,
                 timeout: int = 10):
        self.protocol = protocol
        self.host = host
        self.port = port
        self.timeout = timeout
        self.url = f"{self.protocol}://{self.host}:{self.port}"
        self.response = {"status": 1, "msg": "successful", "data": {}}

    # http://192.168.63.180:8051/master
    def get_kudu_master_info(self):
        res = http("GET", f"{self.url}/masters", timeout=self.timeout)
        if res.status_code != 200:
            return {}

        soup = BeautifulSoup(res.text, 'lxml')
        # 获取页面Live Masters关键字,查找关键字之下table标签,查找table标签内,获取td标签数据
        live_masters = soup.find(text="Live Masters")
        table = live_masters.find_next('table', attrs={'class': 'table-striped'})
        tbody = [tbody.text for tbody in table.find_all('td')]

        master_html = namedtuple("MasterHtml", ["uuid", "role", 'start_time', "registration"])
        items = [master_html(*tbody[i:i + 4])._asdict() for i in range(0, len(tbody), 4)]

        # 正则表达式[\w\.]表示匹配任意一个字母数字下划线或 “.”
        rc = re.compile(
            r'rpc_addresses\s\{\shost:\s\"([\w\.]+)\"\sport:\s(\d+)\s\}\shttp_addresses\s\{\shost:\s\"([\w\.]+)\"\sport:\s(\d+)\s\}')
        result = []
        master_info = namedtuple("MasterInfo", ["uuid", "role", 'rpc_addresses', "http_addresses"])
        for item in items:
            if regex := rc.search(item.get("registration")):
                data = master_info(item.get("uuid"),
                                   item.get("role"),
                                   f"{regex.group(1)}:{regex.group(2)}",
                                   f"{regex.group(3)}:{regex.group(4)}")._asdict()
                result.append(dict(OrderedDict(data)))
        return result

    # http://192.168.63.180:8051/metrics?compact=1&metrics=live_row_count,on_disk_size
    def get_tables_info(self, num_tablet_replicas=3):
        """
        :param num_tablet_replicas: kudu表默认的复制因子是3 http://KUDU_IP:8051/metrics  属性on_disk_size * num_tablet_replicas =总容量
        :return:
        """
        result = []
        table_info = namedtuple("TableInfo", ["table_id", "database_name", "table_name", 'row_count', "raw_capacity", "capacity"])
        try:
            for item in self.get_kudu_master_info():
                if item.get("role").lower() == "leader":
                    self.url = f"http://{item.get('http_addresses')}"
            res = http("GET", f"{self.url}/metrics",
                       params={'compact': 1, 'metrics': 'live_row_count,on_disk_size'}, timeout=self.timeout)
            if res.status_code != 200:
                return result
            for table in loads(res.text):
                # ['impala', 'dw_ods_datalake', 'approval_new_application']
                database_table = re.split(r'[:.]{1,2}', table.get('attributes').get('table_name'))
                if len(database_table) == 3:
                    table_id = table['id']
                    database_name = database_table[1]
                    table_name = database_table[2]
                    row_count = table['metrics'][0]['value']
                    raw_capacity = table['metrics'][1]['value'] * num_tablet_replicas
                    data = table_info(table_id, database_name, table_name, row_count, raw_capacity, unit_convert(raw_capacity))._asdict()
                    result.append(dict(data))
        except Exception as e:
            logger.info(f"get_tables_info error =>{e}")
            return []
        return result

    # http://192.168.63.180:8051/table
    def get_table_partition_num(self, table_id):
        res = http("GET", f"{self.url}/table", params={'id': table_id}, timeout=self.timeout)
        if res.status_code != 200:
            return {}
        soup = BeautifulSoup(res.text, 'lxml')
        summary = soup.find(text="Summary")
        table = summary.find_next('table', attrs={'class': 'table-striped'})
        t_head = [header.text.lower() for header in table.find_all('th')]
        t_head = [header.text.lower() for header in table.find_all('td')]
        return dict(zip(t_head, t_head))


def latest_kudu_table_data(page: int, per_page: int, db: Session):
    data = KuduTableInfo.get_latest_time(db)
    if not data:
        return False
    data = KuduTableInfo.get_latest_time_range(page, per_page, data["create_time"], db)
    if not data:
        return False

    return data


def latest_kudu_db_data(start_time: str, end_time: str, db: Session):
    data = KuduTableInfo.get_db_data_by_date_range(start_time, end_time, db)
    if not data:
        return []
    data = [{'database_name': item['database_name'], 'capacity': unit_convert(item['raw_capacity']), 'date': item['create_date']} for item
            in data]
    return data


def kudu_capacity_report_handler(start_time: str, end_time: str, db: Session):
    try:
        result = latest_kudu_db_data(start_time, end_time, db)
        if not result:
            return [], []
        data = {}
        for item in result:
            if not data.get(item['database_name'], None):
                data[item['database_name']] = {}
                data[item['database_name']].update({item['date']: item['capacity']})
            else:
                data[item['database_name']].update({item['date']: item['capacity']})
        df = pd.DataFrame(data)
        df = df.T
        df.reset_index(drop=False, inplace=True)
        df.rename(columns={"index": "database_name"}, inplace=True)
        data = df.to_dict(orient='split')
        title, content = data['columns'], data['data']
        return title, content
    except Exception as e:
        logger.error(f"capacity_report_handler => {e}")
        return [], []


def kudu_capacity_table_report_handler(start_time: str, end_time: str, db: Session):
    try:
        result = KuduTableInfo.get_db_table_by_date_range(start_time, end_time, db)
        if not result:
            return [], []
        data = {}
        for item in result:
            if not data.get(item['table_name'], None):
                data[item['table_name']] = {}
                data[item['table_name']].update({item['date']: item['capacity']})
            else:
                data[item['table_name']].update({item['date']: item['capacity']})
        df = pd.DataFrame(data)
        df = df.T
        df.reset_index(drop=False, inplace=True)
        df.rename(columns={"index": "table_name"}, inplace=True)
        data = df.to_dict(orient='split')
        title, content = data['columns'], data['data']
        return title, content
    except Exception as e:
        logger.error(f"capacity_kudu_table_report_handler => {e}")
        return [], []


if __name__ == "__main__":
    from pprint import pprint
    from common.database import SessionLocal

    # session = SessionLocal()
    # print(kudu_capacity_report_handler('2023-02-01', '2023-02-07', session))

    # hive_table_info = HiveTableInfo()
    # print(hive_table_info.get_latest_time(session))
    # session.close()

    # kudu_info = KuduInfo(host='172.31.130.34')
    kudu_info = KuduInfo()
    # print(kudu_info.get_kudu_master_info())
    pprint(kudu_info.get_tables_info())
    # print(kudu_info.get_table_partition_num('8468b7d39aa24cb5aae5a0cabe59393b'))
    # pprint(kudu_info.get_kudu_master_info())

    # pprint(kudu_capacity_table_report_handler('2023-02-01', '2023-02-07', session))
